#include "TcpAcceptor.h"
#include "TcpServer.h"

using namespace clusnet::net;

TcpAcceptor::TcpAcceptor(TcpServer* server, boost::asio::io_context& io_context, const boost::asio::ip::tcp::endpoint& endpoint, bool reuseAddr, bool nodelay, bool keepAlive)
    : strand_(io_context),
      io_(io_context),
      acceptor_(io_context, endpoint, reuseAddr),
      newConnCallback_(nullptr),
      closeCallback_(nullptr),
      writeCallback_(nullptr),
      messageCallback_(nullptr),
      nodelay_(nodelay),
      keepAlive_(keepAlive),
      stop_(false),
      server_(server)
{
}

TcpAcceptor::~TcpAcceptor()
{
    assert(!tcpConnPool_.checkLeak());
    acceptor_.close();
}

void TcpAcceptor::safeStartAccept()
{
    boost::asio::dispatch(io_, boost::asio::bind_executor(strand_, [this]()
    {
        if (stop_)
        {
            return;
        }
        assert(server_ != nullptr);

        // release connection, recycle by tcpConnPool, thread safe
        auto conn = tcpConnPool_.acquireWithDelHandler([this](clusnet::base::ObjectPool<TcpConnection>::pointer ptr)
        {
            boost::asio::post(io_, boost::asio::bind_executor(strand_, [this, ptr] { tcpConnPool_.release(ptr); }));
        }, server_->getNextIoContext());
        acceptor_.async_accept(conn->socket(), boost::asio::bind_executor(strand_, boost::bind(&TcpAcceptor::handleAccept, this, conn, boost::asio::placeholders::error)));
    }));
}


void TcpAcceptor::safeStop(const std::function<void()>& cb)
{
    boost::asio::dispatch(io_, boost::asio::bind_executor(strand_, [this, cb]() 
    { 
        stop_ = true;
        acceptor_.cancel();
        cb(); 
    }));
}

void TcpAcceptor::handleAccept(TcpConnectionPtr conn, const boost::system::error_code& ec)
{
    if (!ec)
    {
        conn->initConnect(nodelay_, keepAlive_, writeCallback_, messageCallback_, std::bind(&TcpAcceptor::safeHandleClose, this, std::placeholders::_1));
        newConnCallback_(conn);
    }
    else
    {
        LOG(WARNING) << "accept failed. " << ec.message();
    }
    safeStartAccept();
}

void TcpAcceptor::safeHandleClose(TcpConnectionPtr& conn)
{
    std::weak_ptr<TcpConnection>    ptr = conn;
    boost::asio::dispatch(io_, boost::asio::bind_executor(strand_, [this, ptr]()
    {
        auto conn = ptr.lock();
        assert(conn != nullptr);
        closeCallback_(conn);
    }));
}
